package com.zenitera.bigdata.sink;

import com.alibaba.fastjson.JSON;
import com.zenitera.bigdata.bean.WaterSensor;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;

import java.util.Arrays;
import java.util.List;

public class Flink03_Sink_es02 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);


        SingleOutputStreamOperator<WaterSensor> result = env
                .socketTextStream("hdt-dmcp-ops05", 6666)
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                })
                .keyBy(WaterSensor::getId)
                .sum("vc");

        List<HttpHost> hosts = Arrays.asList(
                new HttpHost("hdt-dmcp-ops01", 9200),
                new HttpHost("hdt-dmcp-ops02", 9200),
                new HttpHost("hdt-dmcp-ops03", 9200)
        );

        ElasticsearchSink.Builder<WaterSensor> builder = new ElasticsearchSink.Builder<WaterSensor>(
                hosts,
                new ElasticsearchSinkFunction<WaterSensor>() {
                    @Override
                    public void process(WaterSensor element,
                                        RuntimeContext ctx,
                                        RequestIndexer indexer) {
                        String msg = JSON.toJSONString(element);

                        IndexRequest ir = Requests
                                .indexRequest("wangttest02")
                                .type("_doc")
                                .id(element.getId())
                                .source(msg, XContentType.JSON);

                        indexer.add(ir);

                    }
                }
        );
        builder.setBulkFlushInterval(2000);
        builder.setBulkFlushMaxSizeMb(1024);
        builder.setBulkFlushMaxActions(2);

        result.addSink(builder.build());

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

/*
[wangting@hdt-dmcp-ops05 ~]$ nc -lk 6666
sensor_1,1,20
sensor_1,1,50
sensor_1,1,50
sensor_1,1,70
sensor_2,2,10
sensor_2,2,30
---------------------
{
"_index": "wangttest02",
"_type": "_doc",
"_id": "sensor_1",
"_version": 4,
"_score": 1,
"_source": {
"id": "sensor_1",
"ts": 1,
"vc": 190
}
}

{
"_index": "wangttest02",
"_type": "_doc",
"_id": "sensor_2",
"_version": 2,
"_score": 1,
"_source": {
"id": "sensor_2",
"ts": 2,
"vc": 40
}
}

 */

